package com.wx.io.multichannel;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多线程版的多路复用器
 * @author wangxun
 * @version 1.0
 * @description: 完成多线程版的多路复用器,多selector
 * @date 2021/1/4 11:37 上午
 */
public class MultiThreads {
    ServerSocketChannel socketChannel = null;
    Selector selector1 =null;
    Selector selector2 =null;
    Selector selector3 =null;
    int port = 9999;


    public void initServer(){
        try {
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            // 非阻塞
            serverChannel.configureBlocking(false);
            serverChannel.bind(new InetSocketAddress(port));
            // 开启多路复用器
             selector1 = Selector.open();
             selector2 = Selector.open();
             selector3 = Selector.open();

            serverChannel.register(selector1, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        MultiThreads multiThreads = new MultiThreads();
        multiThreads.initServer();
        NioThread t1 = new NioThread(multiThreads.selector1, 2);
        NioThread t2 = new NioThread(multiThreads.selector2);
        NioThread t3 = new NioThread(multiThreads.selector3);

        t1.start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t2.start();
        t3.start();
        System.out.println("服务器启动了。。。。。");

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
class NioThread extends Thread{
    Selector selector = null;
    static int selectors =0;
    int id = 0;
    // 接收队列
    volatile  static  BlockingQueue<SocketChannel>[] queues;

    static AtomicInteger idx = new AtomicInteger();

    NioThread(Selector selector,int num){
        this.setName("boss Thread");
        this.selector = selector;
        // num 代表work的selector有几个
        selectors = num;
        queues = new LinkedBlockingQueue[selectors];
        for (int i = 0; i < selectors; i++) {
            queues[i] = new LinkedBlockingQueue<>();
        }
        System.out.println("boss 启动");

    }
    NioThread(Selector selector){
        this.selector = selector;
        id = idx.incrementAndGet() % selectors ;
        this.setName("work Thread"+id);
        System.out.println("work:"+id+" 启动");
    }

    @Override
    public void run() {
        try {
            while(true){
                while (selector.select(10)>0){
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while(iterator.hasNext()){
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        if(key.isAcceptable()){
                            acceptHandler(key);
                        }else if(key.isReadable()){
                            readHandler(key);
                        }
                    }
                }
                // 用队列接收要连接的socket,将通道注册到对应的selector中
                if(!queues[id].isEmpty()){
                    ByteBuffer buffer = ByteBuffer.allocateDirect(8192);
                    SocketChannel client = queues[id].take();
                    client.register(selector,SelectionKey.OP_READ,buffer);
                    System.out.println("-------------------------------------------");
                    System.out.println("新客户端：" + client.socket().getPort()+"分配到："+ this.getName());
                    System.out.println("-------------------------------------------");

                }

            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private void acceptHandler(SelectionKey key) {
        try {
            ServerSocketChannel ssc =  (ServerSocketChannel)key.channel();
            SocketChannel client = ssc.accept();
            client.configureBlocking(false);
            int num =idx.getAndIncrement() % selectors;
            queues[num].add(client);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    private void readHandler(SelectionKey key) {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer buffer =(ByteBuffer) key.attachment();
        buffer.clear();
        int read = 0;
        try {
            while(true){
                read = client.read(buffer);
                if(read>0){
                    buffer.flip();
                    while(buffer.hasRemaining()){
                        client.write(buffer);
                    }
                    buffer.clear();
                }else if(read == 0){
                    break;
                }else{
                    client.close();
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }



}
